/**
 * AggPassiveThread.java
 * ************************** 
 * @date Mar 31, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.aggregation;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.logging.Level;
import java.util.logging.Logger;

import messages.Message;

import gossipServices.aggregation.functions.AggregationUpdateFunction;
import gossipServices.basic.nodeDescriptors.ConcreteNodeDescriptor;
import gossipServices.basic.nodeDescriptors.NodeDescriptor;
import gossipServices.basic.view.PartialView;
import gossipServices.gossipMessages.AggregationMessage;
import gossipServices.gossipThread.PassiveGossipThread;
import gossipServices.network.NetworkServicesAPI;
import gossipServices.pss.PeerSamplingServiceAPI;

/**
 * Description: 
 *
 */
public class AggPassiveThreadOld extends PassiveGossipThread {

	protected PeerSamplingServiceAPI pss;
	protected NetworkServicesAPI nia;
	protected EpochHandler epochHandler;
	protected AggregationUpdateFunction updateFunc;
	protected static SingleValueContainer localValue;
	protected Double initialValue;
	
	private Logger log;
	
	/**
	 * @param name
	 * @param cycleLenghtMsec
	 * @param view
	 * @param function 
	 */
	public AggPassiveThreadOld(String name, long cycleLenghtMsec, PartialView view,
			PeerSamplingServiceAPI pss, NetworkServicesAPI nia, 
			EpochHandler epochHandler,  AggregationUpdateFunction function) {
		super(name, cycleLenghtMsec, view);
		this.pss = pss;
		this.nia = nia;
		this.epochHandler = epochHandler;
		this.updateFunc = function;
		
		log = Logger.getLogger("PassiveAgg-log");
		log.setLevel(Level.OFF);
	}

	@Override
	public void passiveThreadInstructions() throws InterruptedException {
		AggregationMessage msgQ = (AggregationMessage) nia.receive();
		NodeDescriptor me = null;
		try {
			me = new ConcreteNodeDescriptor(nia.getNodeName(),
					InetAddress.getLocalHost(), new Integer(0));
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		
		log.finer("received: "+msgQ);
//		Double sq = msgQ.getValue();
//		log.finer("exchanging");
//		passiveExchange(sq, me, msgQ.getSender());
		
		switch(epochHandler.checkEpochs(msgQ, epochHandler)){
		case exchange:
			log.info("exchanging with "+msgQ.getSender()+"\n before exchange: "+localValue);
			SingleValueDouble sq = (SingleValueDouble) msgQ.getSingleValue();
			passiveExchange(sq.getValue(), me, msgQ.getSender());
			break;

		case sendCorrectEpoch:
			Message corrMsg = epochHandler.composeCorrectionMessage(me, msgQ.getSender(),
					AggActiveThreadOld.class); 
			log.finer("sending correction message "+ corrMsg + " to " + msgQ.getSender() );
			nia.send(corrMsg, msgQ.getSender());
			break;

		case updateCurrentEpoch:
			try {
				log.finer("updating current epoch from "+msgQ);
				epochHandler.updateCurrentEpoch(msgQ);
			} catch (AggregationRestartException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				restart(e.getRemainingCycles());
			}
			break;
		
		default: 
				
		}
			
	}

	/**
	 * 
	 * @param sq
	 * @param me
	 * @param dest
	 */
	private void passiveExchange(Double sq, NodeDescriptor me, NodeDescriptor dest) {
		Message msg;
		synchronized (localValue) {
			msg = new AggregationMessage(me, dest, localValue, AggPassiveThreadOld.class, 
					epochHandler.getEpoch(), epochHandler.getRemainingCycles());			
		}
		nia.send(msg, dest);
		log.finer("sending: "+((AggregationMessage)msg)+" to "+dest);
		synchronized (localValue) {
			localValue.setValue(updateFunc.update(localValue.getValue(), sq.doubleValue()));
			System.out.println(nia.getNodeName()+
					"| Aggregation actual value: "+localValue+" epoch "+
					+epochHandler.getEpoch()+" cycle: "+epochHandler.getElapsedCycles());
		}
		try {
			epochHandler.elapsedCycle();
			log.finer(">>> EPOCH: "+epochHandler.getEpoch()+" cycle: "+epochHandler.getElapsedCycles());
		} catch (AggregationRestartException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			restart(e.getRemainingCycles());
		}
	}

	/**
	 * @throws InterruptedException 
	 * 
	 */
	private void restart(int waitCycles){
		log.finer("restarting");
		synchronized (localValue) {
			localValue.setValue(initialValue);
			System.out.println("Restarting, actual value: "+localValue);
		}
		try {
			if(waitCycles > 0){
				log.finer("sleep "+waitCycles * cycleLenght_msec+" msec");
				Thread.sleep(waitCycles * cycleLenght_msec); //<- may lead to starvation....
				log.finer("wake up!!");
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public final SingleValueContainer getValue() {
		return localValue;
	}

	public final void setValue(SingleValueContainer inValue) {
		localValue = inValue;
	}

	public final Double getInitialValue() {
		return initialValue;
	}

	public final void setInitialValue(Double intialValue) {
		this.initialValue = intialValue;
	}

}
